package edu.gmu.middleware;

import edu.gmu.messages.Client;
import edu.gmu.messages.Connect;
import edu.gmu.messages.Disconnect;
import edu.gmu.messages.Subscribe;
import edu.gmu.messages.Unsubscribe;
import edu.gmu.router.RouterMessage;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * The ClientDispatcher sends Messages to the Sink and
 * receives Messages from the Sink.
 * <p>
 * The dispatcher receives events from the Sink(s) (note 
 * that there could be more than one) and then executes the
 * callback Client's method (i.e. DispatchEvent).
 * <p>
 * The result is that there is only one callback
 * thread  regardless of how many Sink's there are, Client's
 * of the Dispatcher will only have one thread for the callbacks.
 * The consequence is that should the callbacks take a long time
 * to complete, the dispatchQueue may overflow due to too many
 * messages received, though likely an OutOfMemoryError will occur
 * first.
 * <p>
 * The dispatcher also works in the opposite direction by taking a
 * Message object and writing to the Sink - though calls are executed
 * on the caller's thread to include marshaling and writing.
 * 
 * @author  James H. Pope
 * @version $Revision:$ $Date:$
 */
public class ClientDispatcher implements SinkListener
{
    /**
     * The number of byte buffer starting in the pool (may grow?)
     */
    public static final int DEFAULT_INPOOL_SIZE = 5;
    
    /**
     * The size of the byte buffers in the pool - all the same size
     */
    public static final int DEFAULT_INBUFFER_SIZE = 1048576;
    
    private Sink sink                = null;
    private volatile boolean stopped = false;
    
    private Lock lock                = new ReentrantLock(true); // make fair
    private Condition messageReady   = lock.newCondition();
    private List dispatchQueue       = new ArrayList();
    
    private Marshaler marshaler      = new MarshalerImpl();
    private Unmarshaler unmarshaler  = new UnmarshalerImpl();
    
    private ByteBufferPool inPool    = null;
    private ByteBuffer out           = null;
    
    private MessageRegistry registry = new MessageRegistry();
    private List messageListeners    = new ArrayList();
    private List callbackListeners   = new ArrayList();
    
    private Client client            = null;
    
    /**
     * Creates a new ClientDispatcher specified values.  Adds this object
     * as a sinkListener to the sink object.
     * @param sink
     * @param clientId
     * @param clientName
     */
    public ClientDispatcher( Sink sink, int clientId, String clientName )
    {
        this( sink, DEFAULT_INPOOL_SIZE, DEFAULT_INBUFFER_SIZE, clientId, clientName );
    }
    
    /**
     * Creates a new ClientDispatcher specified values.  Adds this object
     * as a sinkListener to the sink object.
     * @param sink
     * @param inPoolSize
     * @param inBufferSize
     * @param clientId
     * @param clientName
     */
    public ClientDispatcher(    Sink sink,
                                int inPoolSize,
                                int inBufferSize,
                                int clientId,
                                String clientName)
    {
        this.sink = sink;
        this.sink.addSinkListener(this);
        
        this.client = new Client();
        this.client.id = clientId;
        this.client.name = clientName;
        
        /*
         * Register RouterMessage bindings
         */
        this.register(RouterMessage.CONNECT,     Connect.class);
        this.register(RouterMessage.DISCONNECT,  Disconnect.class);
        this.register(RouterMessage.SUBSCRIBE,   Subscribe.class);
        this.register(RouterMessage.UNSUBSCRIBE, Unsubscribe.class);
        
        /*
         *  NOTE: BIG ALLOCATIONS HERE
         */
        this.inPool = new ByteBufferPool(inPoolSize,inBufferSize);
        // Write has own buffer so cannopt fail due to lack of buffer
        this.out    = ByteBuffer.allocate( inBufferSize );
    }
    
    /**
     * Registers binding between given id and class
     * @param messageId
     * @param class to bind to messageId
     * @returns the newly registered MessageBinding
     * @throws IllegalArgumentException if id is already bound
     */
    private MessageBinding register( int messageId, Class boundClass )
    {
        return this.registry.register(messageId, boundClass);
    }
    
    /**
     * Sends Subscribe Message to the Router for the indicated messageId.
     * @param topic
     * @throws IllegalArgumentException if topic is already bound
     * @throws java.io.IOException
     */
    public void subscribe( int topic, Class boundClass ) throws IOException
    {
        /*
         * Do not like this design-wise - need to find better solution
         * Currecntly requires client to start sink before calling this
         * This forced order of client method calls is not desireable
         * 
         * Note that adding a message listener does not incur this anomaly???
         */
        //local registration
        this.register(topic, boundClass); 
        
        // remote registration
        //Client client = new Client();
        //client.id = 9876;
        //client.name = "Not specified";
        Subscribe subscribe = new Subscribe();
        subscribe.client = this.client;
        subscribe.topics = new int[]{topic};
        this.send(RouterMessage.SUBSCRIBE, subscribe);
    }
    
    
    /**
     * Adds callback to be executed when specified topic arrives.
     * The callback is passed an instance of the indicated boundClass.
     * @param topic
     * @param class to bind to topic
     * @param callback to be called when id/message is received
     * @returns the newly registered MessageBinding
     * @throws IllegalArgumentException if topic is already bound
     * @throws java.io.IOException
     */
    public void subscribe( int topic, Class boundClass, Callback callback )
            throws IOException
    {
        MessageBinding messageBinding   = this.register(topic, boundClass);
        CallbackBinding callbackBinding = new CallbackBinding(  messageBinding,
                                                                callback);
        this.callbackListeners.add( callbackBinding );
    }
    
    /**
     * Sends Unsubscribe Message to the Router for the indicated topic.
     * @param topic
     * @throws java.io.IOException
     */
    public void unsubscribe( int topic ) throws IOException
    {
        /*
         * Do not like this design-wise - need to find better solution
         * Currecntly requires client to start sink before calling this
         * This forced order of client method calls is not desireable
         * 
         * Note that adding a message listener does not incur this anomaly???
         */
        //local unregister
        this.registry.unregister(topic); 
        
        // remote unregister
        Unsubscribe unsubscribe = new Unsubscribe();
        unsubscribe.client = this.client;
        unsubscribe.topics = new int[]{topic};
        this.send(RouterMessage.UNSUBSCRIBE, unsubscribe);
    }
    
    /**
     * Marshals message and writes to sink - both are carried out on
     * the caller's thread.
     * @param topic
     * @param message
     * @throws IOException if the unmashaling/write process was not successful
     */
    public void publish( int topic, Message message ) throws IOException
    {
        this.send(topic, message);
    }
    
    /**
     * Marshals message and writes to sink - both are carried out on
     * the caller's thread.
     * @param topic
     * @param message
     * @throws IOException if the unmashaling/write process was not successful
     */
    private void send( int topic, Message message ) throws IOException
    {
        Header header = new Header();
        header.setTopic(topic);
        this.marshaler.marshal(header, message, out);
        //System.out.print("Writing...");
        //print(out,12);
        this.sink.write(out);
    }
    
    
    
    /**
     * Called from source object when the sink is disconnected.
     * @param event
     */
    public void sinkDisconnected( SinkDisconnectedEvent event )
    {
        // Regardless of disconnect type - go ahead and shutdown
        // Another option is if exception, try to reconnect
        //--------------------------------------------------------------------//
        // Called on Sink readerThread - last statement it executes
        //--------------------------------------------------------------------//
        this.stop();
        //--------------------------------------------------------------------//
    }
    
    
    /**
     * Called from source object when message is received.
     * @param event
     * @throws IOException if no free buffer to copy data into
     */
    public void dataReceived( SinkDataEvent event )
    {
        //--------------------------------------------------------------------//
        // Note that we copy on caller's thread and unmarshal on dispatchThread
        //--------------------------------------------------------------------//
        /*
         * Make copy and return as quickly as possible.
         */
        // NEED TO REASON - WHY FAILES WHEN THIS IS INSIDE LOCK????
        ByteBuffer in = event.getIn();

        ByteBuffer inBuffer = this.inPool.checkout();
        if( inBuffer == null )
        {
            String s = "No free buffer in inPool to unmarshal message - dropping.";
            System.out.println( s );
            this.stop();
            //IOException e = new IOException(s);
            //throw e;
            //this.fireExceptionReceived(e);
            return;
        }

        /*
         * We have a free buffer - go ahead and copy in to the inPool
         * buffer - we will unmarshal on a different thread
         * after which the inPool will be returned to the pool
         */
        inBuffer.clear();
        inBuffer.put(in);
        inBuffer.flip();

        this.lock.lock();
        try
        {
            this.dispatchQueue.add( new DispatchEntry(  event.getHeader(),
                                                        inBuffer,
                                                        event.getSink() ) );
            this.messageReady.signal();
        }
        finally
        {
            this.lock.unlock();
        }
        //--------------------------------------------------------------------//
    }
    
    
    
    /**
     * Starts a new thread an reads continuously from the sink firing
     * events when packets are received/unmarshaled.
     */
    public void start( )
    {
        Runnable runnable = new Runnable()
        {
            public void run()
            {
                while( stopped == false )
                {
                    try
                    {
                        List tempDispatchQueue = new ArrayList();
                        lock.lock();
                        try
                        {
                            // Make sure we have not received more
                            if( dispatchQueue.size() == 0 )
                            {
                                // Blocks until a message has been added to queue
                                messageReady.await();
                            }
                            
                            // Make copy so that queue can fill while callbacks
                            tempDispatchQueue.addAll(dispatchQueue);
                            dispatchQueue.clear();
                        }
                        finally
                        {
                            lock.unlock();
                        }
                        
                        // Empty the queue executing callbacks
                        //System.out.println( "    Queue size "+tempDispatchQueue.size() );
                        for( int i = 0; i < tempDispatchQueue.size(); i++ )
                        {
                            // FIFO order?
                            DispatchEntry entry = (DispatchEntry)tempDispatchQueue.get(i);
                            ByteBuffer inBuffer = entry.getInBuffer();
                            Header header       = entry.getHeader();
                            
                            //System.out.println("Client received...");
                            //print( inBuffer, 12 );
                            
                            int topic = header.getTopic();
                            MessageBinding messageBinding = registry.getMessageBinding(topic);
                            if( messageBinding != null )
                            {
                                // This will takes a while
                                Message message = unmarshaler.unmarshal(header,
                                                                        inBuffer,
                                                                        messageBinding);
                                /*
                                 * Note the order
                                 * 
                                 * 1. Call messageListeners
                                 * 2. Call callbackListeners
                                 */
                                // Notify messageListeners
                                fireMessageReceived( header, message );

                                // Nofity callbackListners
                                fireCallbacks( header, message );
                            }
                            else
                            {
                                System.out.println( "No messageBinding for topic "+topic );
                            }
                            
                            inPool.checkin(inBuffer);
                        }
                    }
                    catch( InterruptedException e )
                    {
                        // Catch condition variable exception - not normal
                        //fireExceptionReceived(e);
                        e.printStackTrace();
                    }
                    catch( Exception e )
                    {
                        // Catch user callback exceptions
                        //fireExceptionReceived(e);
                        e.printStackTrace();
                    }
                }
                System.out.println("###################################");
                System.out.println("# Client Dispatcher exitting - bye!");
                System.out.println("###################################");
            }
        };
        Thread thread = new Thread(runnable);
        //thread.setDaemon(true);
        thread.setName( "DispatcherThread" );
        thread.start();
    }
    
    
    /**
     * Stops the ClientDispatcher from sending/receiving. Also results in
     * the sink being stopped.
     */
    public void stop( )
    {
        this.stopped = true;
        lock.lock();
        try
        {
            this.sink.disconnect();
            this.messageReady.signal();
        }
        finally
        {
            lock.unlock();
        }
        
    }
    
    /**
     * Notify all listeners that have registered interest for notification
     * on this event.
     * @param header
     * @param message
     * @param messageBinding
     */
    public void fireCallbacks(  Header header,
                                Message message )
    {
        //Notify listeners last to first, notifying those that are interested
        for(int i = 0; i < this.callbackListeners.size(); i++)
        {
            CallbackBinding callbackBinding =
                    (CallbackBinding)this.callbackListeners.get(i);
            
            MessageBinding rMessageBinding =
                    callbackBinding.getMessageBinding();
            
            if( rMessageBinding.getTopic() == header.getTopic() )
            {
                Callback callback = callbackBinding.getCallback();
                callback.execute(message);
            }
        }
    }
    
    
    /**
     * Notify all listeners that have registered interest for notification
     * on this event.
     * @param header
     * @param message
     */
    public void fireMessageReceived( Header header, Message message )
    {
        //Do this in order to only create Event once
        MessageEvent event = null;

        //Notify listeners last to first, notifying those that are interested
        for(int i = 0; i < this.messageListeners.size(); i++)
        {
            // lazy initialization
            if(event == null)
            {
                event = new MessageEvent(   this,
                                            (long)System.currentTimeMillis(),
                                            header,
                                            message
                                         );
            }
            // Get listener
            MessageListener listener = ((MessageListener)this.messageListeners.get(i));
            // Call listener
            listener.messageReceived( event );
        }
    }
    
    /**
     * Adds to list of listeners. First checks to see if
     * listener is already in list, if it is it does not add it a second time.
     * @param listener to add.
     */
    public void addMessageListener( MessageListener listener )
    {
        if( !this.messageListeners.contains(listener) )
        {
            this.messageListeners.add(listener);
        }
    }

    /**
     * Removes from list of listeners.
     * @param listener to remove.
     */
    public void removeMessageListener( MessageListener listener )
    {
        this.messageListeners.remove(listener);
    }
    
    
    
    /**
     * Solely to be used for debugging.
     * @param inBuffer
     * @param length
     */
    private static void print( ByteBuffer inBuffer, int length )
    {
        byte[] in = inBuffer.array();
        System.out.print( "position="+inBuffer.position() + " limit="+inBuffer.limit() );
        for (int j = 0; j < length; j++)
        {
            String s = Integer.toHexString( in[j]&0xff );
            if( s.length() == 1 ) s = "0"+s;
            System.out.print( " " + s );
        }
        System.out.println( "" );
    }
    
    


}
